package com.liaofan.rocketmq.batch;

import org.apache.rocketmq.common.message.Message;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

//消息列表分割器
public class MessageListSplitter implements Iterator<List<Message>> {
    //指定极限值是4M
    private final int SIZE_LIMIT = 4 * 1024 * 1024;
    private final List<Message> messages;
    //小集合的起始索引
    private int currentIndex;

    public MessageListSplitter(List<Message> messageList) {
        this.messages = messageList;
    }


    @Override
    public boolean hasNext() {
        //判断当前便利的消息索引小于消息总数
        return currentIndex < messages.size();
    }

    @Override
    public List<Message> next() {
        int nextIndex = currentIndex;
        int totalSize = 0;

        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            //统计当前遍历的message的大小
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            tmpSize = tmpSize + 20;

            if (tmpSize > SIZE_LIMIT) {
                if (nextIndex - currentIndex == 0) {
                    nextIndex++;
                }

                break;
            }

            if (tmpSize + totalSize > SIZE_LIMIT) {
                break;
            } else {
                totalSize += tmpSize;
            }
        }

        List<Message> subList = messages.subList(currentIndex, nextIndex);
        currentIndex = nextIndex;
        return subList;
    }
}
